package com.bwie.realtime.jtp.dwd.log.function;

import com.bwie.realtime.jtp.dwd.log.bean.PageViewBean;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewReportWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {
    @Override
    public void apply(String key, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception {
//        窗口时间
        String windowStart  =  DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEnd  =  DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");

//        获取窗口中计算结果
        PageViewBean bean =   input.iterator().next();
        Long pvCount = bean.getPvCount();
        Long pvDuringTime = bean.getPvDuringTime();
        Long uvCount = bean.getUvCount();
        Long sessionCount = bean.getSessionCount();

//        输出
        String output=windowStart+","+windowEnd+","
                +key+","
                +pvCount+","+pvDuringTime+","+uvCount+","+sessionCount+","
                +System.currentTimeMillis();
        out.collect(output);
    }
}
